package tk.xboot.flink.vs;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.util.JSONPObject;
import scala.util.parsing.json.JSONObject;


public class VsFun extends RichMapFunction<Tuple2<String, Double>, Tuple2<String, Double>> {

    Object keyedState;

    @Override
    public void open(Configuration parameters) throws Exception {
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.seconds(864000))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
        ValueStateDescriptor<JSONObject> descriptor = new ValueStateDescriptor("keyedstate", TypeInformation.of(JSONPObject.class));

        descriptor.enableTimeToLive(ttlConfig);     // 加入后error
        keyedState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public Tuple2<String, Double> map(Tuple2<String, Double> stringDoubleTuple2) throws Exception {
        return null;
    }
}
